package cn.doitedu.api;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;


/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/2/28
 * @Desc: 学大数据，上多易教育
 *
 *     join算子演示
 **/
public class _31_Join_Demo {

    public static void main(String[] args) throws Exception {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        // {"order_id":"o1","order_amount":3000}
        DataStream<OrderMain> stream1 = env.
                socketTextStream("doitedu", 8891)
                .map(s -> JSON.parseObject(s, OrderMain.class));

        // o1,p1,1200
        // o1,p2,1800
        DataStream<OrderItem> stream2 = env
                .socketTextStream("doitedu", 8892)
                .map(s -> JSON.parseObject(s, OrderItem.class));

        // 两个流的数据关联
        stream1.join(stream2)
                .where(OrderMain::getOrderId)  // 指定流1中的关联条件字段
                .equalTo(OrderItem::getOrderId) // 等于 流2中的关联条件字段
                .window(TumblingProcessingTimeWindows.of(Time.minutes(1)))
                .apply(new JoinFunction<OrderMain, OrderItem, String>() {
                    @Override
                    public String join(OrderMain first, OrderItem second) throws Exception {

                        System.out.println("我被调用了.......................");
                        JSONObject res = new JSONObject();
                        res.put("order_id",first.getOrderId());
                        res.put("order_amount",first.getOrderAmount());
                        res.put("product_id",second.getProductId());
                        res.put("product_amount",second.getProductAmount());

                        return res.toJSONString();
                    }
                }).print();



        env.execute();



    }
}

@Data
@NoArgsConstructor
@AllArgsConstructor
class OrderMain {
    private String orderId;
    private double orderAmount;
}

@Data
@NoArgsConstructor
@AllArgsConstructor
class OrderItem {
    private String orderId;
    private String productId;
    private double productAmount;
}
